package com.shujia.core

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimerService
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo07Fraud {
  def main(args: Array[String]): Unit = {
    /**
     * 1001,0.1
     * 1001,1000
     * 1001,0.01
     * 1001,100
     * 1001,1000
     * 1001,0.1
     * 1001,0.2
     * 1001,1000
     * 1001,1100
     * 1001,10000
     * 1001,20000
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val dealDS: DataStream[(String, Double)] = env.socketTextStream("master", 8888)
      .map(deal => {
        val dealArr: Array[String] = deal.split(",")
        (dealArr(0), dealArr(1).toDouble)
      })

    /**
     * 判定欺诈发生的条件：
     * 1、上一笔交易小于1元
     * 2、当前交易大于500元
     * 3、如果两笔交易之间超过一个小时 则需要重新判断
     * 返回： 用户id,上一笔交易的金额,当前交易的金额
     * (String,Double,Double)
     */
    dealDS
      // 按照每个用户进行分组
      .keyBy(_._1)
      .process(new KeyedProcessFunction[String, (String, Double), (String, Double, Double)] {
        var lastAmountValueState: ValueState[Double] = _

        override def open(parameters: Configuration): Unit = {

          val ctx: RuntimeContext = getRuntimeContext
          val lastAmountValueStateDesc = new ValueStateDescriptor[Double]("lastAmount", classOf[Double])
          lastAmountValueState = ctx.getState(lastAmountValueStateDesc)
        }

        override def processElement(value: (String, Double),
                                    ctx: KeyedProcessFunction[String, (String, Double), (String, Double, Double)]#Context,
                                    out: Collector[(String, Double, Double)]): Unit = {
          // value 表示当前接收到的交易行为数据
          val (userId, amount): (String, Double) = value

          // 获取上一笔交易的金额
          val lastAmount: Double = lastAmountValueState.value()
          // lastAmountValueState.value() != 0  为了避免数据第一次进来 导致误判
          // 还可以让定时器任务重新初始化判断
          if (lastAmountValueState.value() != 0 & lastAmount < 1) {
            // 如果当前交易金额大于500元则触发警告
            if (amount > 500) {
              out.collect((userId, lastAmount, amount))
            }
          }
          lastAmountValueState.update(amount)

          // 获取当前的时间
          val timerService: TimerService = ctx.timerService()
          val currentTS: Long = timerService.currentProcessingTime()
          // 注册一个定时器 指定一个时间间隔 当达到该时间间隔就会触发定时器任务
          // 如果在任务中多次注册定时器 只会刷新定时器的时间 不会导致出现多个定时器
          timerService.registerProcessingTimeTimer(currentTS + 60 * 60 * 1000)
        }

        // 定时器任务
        override def onTimer(timestamp: Long,
                             ctx: KeyedProcessFunction[String, (String, Double), (String, Double, Double)]#OnTimerContext,
                             out: Collector[(String, Double, Double)]): Unit = {
          lastAmountValueState.update(0)

        }
      })
      .print()

    env.execute()


  }

}
